package mr

import (
	"errors"
	"fmt"
	"log"
	"strconv"
	"sync"
	"time"
)
import "net"
import "os"
import "net/rpc"
import "net/http"

type Coordinator struct {
	IsMapPhase  bool
	AllFinished bool
	Lock        sync.Mutex
	MapTasks    []Task
	ReduceTasks []Task
}

const TimeoutSecond = 10

// Your code here -- RPC handlers for the worker to call.

func (c *Coordinator) ReportDone(args *ReportDoneArgs, reply *ReportDoneReply) error {
	c.Lock.Lock()
	defer c.Lock.Unlock()

	fmt.Printf("Task %+v is finished\n\n", args.Task)

	task := args.Task
	if task.Type == Mapper {
		for i := range c.MapTasks {
			if c.MapTasks[i].ID == task.ID {
				c.MapTasks[i].IsDone = true
				return nil
			}
		}
	} else if task.Type == Reducer {
		for i := range c.ReduceTasks {
			if c.ReduceTasks[i].ID == task.ID {
				c.ReduceTasks[i].IsDone = true
				return nil
			}
		}
	}
	return errors.New("not support such task type")
}

func (c *Coordinator) DistributeTask(args *GetTaskArgs, reply *GetTaskReply) error {
	c.Lock.Lock()
	defer c.Lock.Unlock()

	allDone := true
	var tasks []Task

	if c.IsMapPhase {
		tasks = c.MapTasks
	} else {
		tasks = c.ReduceTasks
	}

	for i := range tasks {
		task := &tasks[i] // 通过指针
		if !task.IsDone {
			allDone = false
		}

		unix := time.Now().Unix()
		if task.IsDone || unix-task.StartTime < TimeoutSecond {
			continue
		}

		task.StartTime = unix
		reply.Task = task
		return nil
	}

	if allDone {
		if c.IsMapPhase {
			// 全部结束，进入 Reduce 阶段
			fmt.Println("Map tasks finished, go reduce...")
			c.IsMapPhase = false
			// return c.DistributeTask(args, reply) 本来是递归重试的，奈何 Go 生态太差了，楞是没找到可重入锁
			reply.Task = &Task{Type: Sleep}
		} else {
			// 全部结束，宣告完成，Worker 退出
			fmt.Println("Reduce tasks finished, all finished")
			c.AllFinished = true
			reply.Task = &Task{Type: Exit}
		}
	} else {
		// 还有任务没结束，让 Worker 等待
		reply.Task = &Task{Type: Sleep}
	}
	return nil
}

func (c *Coordinator) Example(args *ExampleArgs, reply *ExampleReply) error {
	reply.Y = args.X + 1
	return nil
}

// start a thread that listens for RPCs from worker.go
func (c *Coordinator) server() {
	rpc.Register(c)
	rpc.HandleHTTP()
	//l, e := net.Listen("tcp", ":1234")
	sockname := coordinatorSock()
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

func (c *Coordinator) Done() bool {
	return c.AllFinished
}

func MakeCoordinator(files []string, nReduce int) *Coordinator {
	// Your code here.
	mapTasks := make([]Task, len(files))
	for i := range files {
		mapTasks[i] = Task{
			Type:      Mapper,
			IsDone:    false,
			InputFile: files[i],
			ID:        i,
			NReduce:   nReduce,
			StartTime: -1,
		}
	}

	reduceTasks := make([]Task, nReduce)
	for i := 0; i < nReduce; i++ {
		reduceTasks[i] = Task{
			Type:      Reducer,
			IsDone:    false,
			InputFile: "mr-*-" + strconv.Itoa(i),
			ID:        i,
			NReduce:   nReduce,
			StartTime: -1,
		}
	}

	c := Coordinator{
		MapTasks:    mapTasks,
		ReduceTasks: reduceTasks,
		IsMapPhase:  true,
		AllFinished: false,
		Lock:        sync.Mutex{},
	}

	c.server()
	return &c
}
